package com.song.hadoop01;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;
import org.slf4j.LoggerFactory;

import com.song.hadoop01.WordCount.WordCountMap;
import com.song.hadoop01.WordCount.wordCountReduce;

public class JoinMapperReduce extends Configured implements Tool {

	private static final  Logger logger = Logger.getLogger(JoinMapperReduce.class);
	
	private static class JoinMapper extends Mapper<LongWritable, Text, IntWritable, Text>
	{
		
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			
			String path = ((FileSplit)context.getInputSplit()).getPath().toString();
			System.out.println("path is"+ path);
			String tag="";
			
			if(path.contains("tablea"))
			{
				tag ="A";
			}else{
				tag ="B";
			}
			if(null==value)
				return;
			if(!value.toString().contains("\t"))
				return ;
			int pos = value.toString().indexOf("\t");
			String keystr = value.toString().substring(0, pos);
			int keyval = Integer.parseInt(keystr.trim());
			String valstr = value.toString().substring(pos+1);
			valstr = tag+"#"+valstr;
			logger.info("keystr is"+keystr+"valstr is"+valstr);
			context.write(new IntWritable(keyval),new Text(valstr));
		}
		
	}
	
	private static class JoinReduce extends Reducer<IntWritable, Text, IntWritable, Text>
	{
		
		
		protected void reduce(IntWritable key, Iterable<Text> items, Context context)
				throws IOException, InterruptedException {
			
			List<String> left = new ArrayList<String>();
			List<String> right = new ArrayList<String>();
			Iterator it = items.iterator();
			while(it.hasNext())
			{
				String item = it.next().toString();
				String val  =item.substring(item.indexOf("#")+1);
				if(item.startsWith("A"))
				{
					left.add(val);
				}else{
					right.add(val);
				}
			}
		
			for(String le :left)
			{
				for(String ri:right)
				{
					
					context.write(key, new Text(le+"\t"+ri+"\n")); // 每调用一次就会生成一行
				}
			}
			

			
		}


		
	}
	
	
	public int run(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = new Job(conf);
		job.setJarByClass(JoinMapperReduce.class);
		job.setJobName(" JoinMapperReduce ");
		
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(Text.class);
		
		job.setMapperClass(JoinMapper.class);
		job.setReducerClass(JoinReduce.class);
		
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		job.waitForCompletion(true);
		return job.isSuccessful()? 1:0;
	}
	
	
	public static void main(String[] args) {
		try {
			int ret = ToolRunner.run(new JoinMapperReduce(), args);
			
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
